
package com.etlmaster.executor.bigdata.hbase;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;

import org.apache.commons.lang3.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.KeyValue;
import org.apache.hadoop.hbase.io.ImmutableBytesWritable;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
public class HFileGenerateMapper extends Mapper<LongWritable, Text, ImmutableBytesWritable, KeyValue>
{
	private String fileFieldName;
	private String fileSeparator;
	private String []columnFamily;
	private String rowKeyColumn;

	private int[] rowKeyIndexs;
	@Override
	protected void setup(Context context) throws IOException, InterruptedException
	{
		Configuration conf = context.getConfiguration();
		this.columnFamily = conf.get("columnFamily").split(",");
		this.rowKeyColumn = conf.get("rowKeyColumn");
		this.fileFieldName = conf.get("fileFieldName");
		
		this.fileSeparator = conf.get("fileSeparator");
		String[] columnsArr = StringUtils.splitByWholeSeparatorPreserveAllTokens(fileFieldName, ",");
		
		String[] rowKeys = StringUtils.splitByWholeSeparatorPreserveAllTokens(rowKeyColumn, ",");
		rowKeyIndexs = new int[rowKeys.length];
		for (int j = 0; j < rowKeys.length; j++)
		{
			for (int i = 0; i < columnsArr.length; i++)
			{
				if (rowKeys[j].equalsIgnoreCase(columnsArr[i]))
				{
					rowKeyIndexs[j] = i;
				}
			}
		}
	}
	

	@Override
	protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException
	{
		
		String[] values = null;
		values = StringUtils.splitByWholeSeparatorPreserveAllTokens(value.toString(), fileSeparator);
		StringBuilder builder = new StringBuilder();
		for (int i = 0; i < rowKeyIndexs.length; i++)
		{
			builder.append(values[rowKeyIndexs[i]].trim());
		}
		String rowKeyStr = builder.toString();
		ImmutableBytesWritable rowkey = new ImmutableBytesWritable(rowKeyStr.getBytes());
		List<KeyValue> list = createKeyValue(rowKeyStr, values);
		Iterator<KeyValue> it = list.iterator();
		while (it.hasNext())
		{
			KeyValue kv = new KeyValue();
			kv = it.next();
			if (kv != null)
			{
				context.write(rowkey, kv);
			}
		}
	}

	private List<KeyValue> createKeyValue(String rowKeyStr, String[] values)
	{
		String[] fileFields = StringUtils.splitByWholeSeparatorPreserveAllTokens(fileFieldName, ",");
		List<KeyValue> list = new ArrayList<KeyValue>(fileFields.length);
		long timeStamp = System.currentTimeMillis();
		for (int j = 0; j < fileFields.length; j++)
		{
			KeyValue kv = new KeyValue(Bytes.toBytes(rowKeyStr), Bytes.toBytes(columnFamily[j]), Bytes.toBytes(fileFields[j]), timeStamp,
					Bytes.toBytes(values[j]));
			list.add(kv);
		}
		return list;
	}
}
